package main

import (
	"fmt"
	"sync"

	"github.com/sirupsen/logrus"

	"logproject/conf"
	"logproject/etcd"
	"logproject/kafka"
	"logproject/tailfile"
	"logproject/utils"
)

//logagent 日志收集客户端
//收集指定目录下的日志文件，发送到kafaka中

//sarama往kafka里发msg
//tailf包读日志文件

func main() {
	//获取本机IP
	ip, err := utils.GetLocalIP()
	if err != nil {
		logrus.Error(ip)
		return
	}
	//根据ip获取etcd中的配置
	configObj := conf.NewConfig().CreateConf()
	configObj.EtcdConfig.CollectKey = fmt.Sprintf(configObj.EtcdConfig.CollectKey, ip)
	etcdobj, err := etcd.Init([]string{configObj.EtcdConfig.Address})
	if err != nil {
		return
	}
	logrus.Info("etcd init success")
	//拉取etcd中文件夹配置
	confList := etcdobj.GetConf(configObj.EtcdConfig.CollectKey)
	// log->taifANDkafka init->tailf读出 ->sarama -> kafka
	kafkaobj, err, registerpcs := kafka.Init([]string{configObj.KafkaConfig.Address}, configObj.KafkaConfig.ChanSize, confList)
	if err != nil {
		return
	}
	logrus.Info("init kafka success")
	// 根据配置文件tail收集日志
	wg := sync.WaitGroup{}
	wg.Add(1)
	registerTasks, tailfstructForSave, err := tailfile.Init(confList, kafkaobj)
	if err != nil {
		wg.Done()
		return
	}
	logrus.Info("init tailf success")
	//得到后 继续监控etcd中配置的变化
	go etcdobj.WatchConf(configObj.EtcdConfig.CollectKey, tailfstructForSave)
	//随配置文件变动管理task
	go tailfile.ChangeTailTasks(tailfstructForSave, registerTasks, kafkaobj, registerpcs)
	wg.Wait()
}
